package com.yox.rpc.serializer;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.yox.rpc.entity.RpcRequest;
import com.yox.rpc.entity.RpcResponse;
import com.yox.rpc.enumeration.SerializerCode;
import com.yox.rpc.exception.SerializeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

/**
 * Kryo 序列化器
 * Kryo 使用Output和Input类完成数据的输入和输出，但是这两个类并不是线程安全的，所以使用ThreadLocal解决线程安全问题
 * 在序列化时，先创建一个Output对象，接着使用writeObject()方法将对象写入Output中，最后调用Output对象的toByte()方法获得字节数组
 * 反序列化则是从Input对象中直接readObject，这里只需要传入对象的类型，而不需要具体传入每个属性的类型信息。
 */
public class KryoSerializer implements CommonSerializer{

    private static final Logger logger = LoggerFactory.getLogger(KryoSerializer.class);

    /**
     * 采用 ThreadLocal 解决线程安全问题
     * 当Kryo写出一个对象实例的时候，需要写出一些标识对象类的东西
     * 这里采用注册类的方式，它将该类与一个int类型的ID相关联，当Kryo写出 RpcRequest 的一个实例时，它会写出这个ID，更高效
     */
    private static final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        // 注册类
        kryo.register(RpcRequest.class);
        kryo.register(RpcResponse.class);
        // 循环引用检测，默认为true
        kryo.setReferences(true);
        // 不强制要求注册类，默认为false，若设置为true则要求涉及到的所有类都要注册，包括jdk中的Object
        kryo.setRegistrationRequired(false);
        return kryo;
    });


    @Override
    public byte[] serialize(Object obj) {
        try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
             Output output = new Output(byteArrayOutputStream)){
            Kryo kryo = kryoThreadLocal.get();
            kryo.writeObject(output, obj);
            kryoThreadLocal.remove();
            return output.toBytes();
        } catch (Exception e) {
            logger.error("序列化时有错误发生", e);
            throw new SerializeException("序列化时发生错误");
        }
    }

    @Override
    public Object deserialize(byte[] bytes, Class<?> clazz) {
        try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
             Input input = new Input(byteArrayInputStream)){
            Kryo kryo = kryoThreadLocal.get();
            Object object = kryo.readObject(input, clazz);
            kryoThreadLocal.remove();
            return object;
        } catch (Exception e) {
            logger.error("反序列化时发生错误", e);
            throw new SerializeException("反序列化时发生错误");
        }
    }

    @Override
    public int getCode() {
        return SerializerCode.valueOf("KRYO").getCode();
    }
}
